#define _MULTI_THREADED
#include <pthread.h>
#include <sys/socket.h>
#include <string.h>     /* for memset() */
#include <arpa/inet.h>  /* for sockaddr_in and inet_ntoa() */
#include <stdio.h>
#include <stdlib.h>     /* for atoi() and exit() */
#include <unistd.h>     /* for close() */
#include <time.h>
#include <sys/time.h>
#include <math.h>
#include <linux/if_ether.h>
#include <netinet/tcp.h>
#include <linux/ip.h>
#include <iostream>
#include <sys/stat.h>
#include <sched.h>
#include "ntcp.h"


struct itimerspec its_da, its_dsack, its_rt_data, its_empty, its_tp;
sigset_t mask_da, mask_dsack, mask_rt_data, mask_tp;
struct sigaction sa_da, sa_dsack, sa_rt_data, sa_tp;

bool log_permit = false;

using namespace std;

void DieWithError(string errorMessage)  /* Error handling function */
{
	cout<<"Error Message: "<<errorMessage<<endl;
	exit(-1);
}

int
int_max(int a, int b){
	return (a>b)?a:b;
}

int
int_min(int a, int b){
	return (a<b)?a:b;
}

double
double_max(double a, double b){
	return (a>b)?a:b;
}

double
get_time(struct ntcp_sock * sock) {
	struct timeval now;
	gettimeofday(&now, NULL);
	double time = (now.tv_sec - sock->start_time.tv_sec) + (now.tv_usec - sock->start_time.tv_usec)/1000000.0;
	return time;
}


void
ntcp_init_timer(struct ntcp_sock* sock)
{
	sa_da.sa_flags = SA_SIGINFO;
	sa_da.sa_sigaction = ntcp_sendAck_handler;
	sigemptyset(&sa_da.sa_mask);

	sa_dsack.sa_flags = SA_SIGINFO;
	sa_dsack.sa_sigaction = ntcp_sendSack_handler;
	sigemptyset(&sa_dsack.sa_mask);

	sa_rt_data.sa_flags = SA_SIGINFO;
	sa_rt_data.sa_sigaction = ntcp_retransmitData_handler;
	sigemptyset(&sa_rt_data.sa_mask);

	sa_tp.sa_flags = SA_SIGINFO;
	sa_tp.sa_sigaction = ntcp_print_throughput_handler;
	sigemptyset(&sa_tp.sa_mask);

	if(sigaction(SIG_DELAY_ACK, &sa_da, NULL) == -1)
		DieWithError("delay ack sigaction failed");

	if(sigaction(SIG_DELAY_SACK, &sa_dsack, NULL) == -1)
		DieWithError("retransmit ack sigaction failed");

	if(sigaction(SIG_RETRANSMIT_DATA, &sa_rt_data, NULL) == -1)
		DieWithError("retransmit data sigaction failed");		
	
	if(sigaction(SIG_THROUGHPUT, &sa_tp, NULL) == -1)
		DieWithError("print throughput sigaction failed");

	sigemptyset(&mask_da);
	sigemptyset(&mask_dsack);
	sigemptyset(&mask_rt_data);
	sigemptyset(&mask_tp);
	sigaddset(&mask_da, SIG_DELAY_ACK);
	sigaddset(&mask_dsack, SIG_DELAY_SACK);
	sigaddset(&mask_rt_data, SIG_RETRANSMIT_DATA);
	sigaddset(&mask_tp, SIG_THROUGHPUT);

	sock->sev_da.sigev_notify = SIGEV_SIGNAL;
	sock->sev_da.sigev_signo = SIG_DELAY_ACK;
	sock->sev_da.sigev_value.sival_ptr = sock;

	if(timer_create(CLOCKID, &sock->sev_da, &sock->timer_da) == -1)
		DieWithError("delay ack timer create failed");

	
	sock->sev_dsack.sigev_notify = SIGEV_SIGNAL;
	sock->sev_dsack.sigev_signo = SIG_DELAY_SACK;
	sock->sev_dsack.sigev_value.sival_ptr = sock;
	
	if(timer_create(CLOCKID, &sock->sev_dsack, &sock->timer_dsack) == -1)
		DieWithError("delay SACK timer create failed");
	
	sock->sev_rt_data.sigev_notify = SIGEV_SIGNAL;
	sock->sev_rt_data.sigev_signo = SIG_RETRANSMIT_DATA;
	sock->sev_rt_data.sigev_value.sival_ptr = sock;
	
	if(timer_create(CLOCKID, &sock->sev_rt_data, &sock->timer_rt_data) == -1)
		DieWithError("retransmit DATA timer create failed");


	sock->sev_tp.sigev_notify = SIGEV_SIGNAL;
	sock->sev_tp.sigev_signo = SIG_THROUGHPUT;
	sock->sev_tp.sigev_value.sival_ptr = sock;

	if(timer_create(CLOCKID, &sock->sev_tp, &sock->timer_tp) == -1)
		DieWithError("print throughput timer create failed");

	//timer no loop
	its_da.it_value.tv_sec = 0;
	its_da.it_value.tv_nsec = 40000000;
	its_da.it_interval.tv_sec = 0;
	its_da.it_interval.tv_nsec = 0;

	its_dsack.it_value.tv_sec = 0;
	its_dsack.it_value.tv_nsec = 40000000;
	its_dsack.it_interval.tv_sec = 0;
	its_dsack.it_interval.tv_nsec = 0;
	
	its_rt_data.it_value.tv_sec = 0;
	its_rt_data.it_value.tv_nsec = 200000000;
	its_rt_data.it_interval.tv_sec = 0;
	its_rt_data.it_interval.tv_nsec = 0;
	
	its_empty.it_value.tv_sec = 0;
	its_empty.it_value.tv_nsec = 0;
	its_empty.it_interval.tv_sec = 0;
	its_empty.it_interval.tv_nsec = 0;

	its_tp.it_value.tv_sec = 0;
	its_tp.it_value.tv_nsec = 1000000 * SAMPLE_INV;
	its_tp.it_interval.tv_sec = 0;
	its_tp.it_interval.tv_nsec = 1000000 * SAMPLE_INV;
}


void
ntcp_init_threads(struct ntcp_sock* sock)
{
	if(sock->sendThread == -1){
    		if( pthread_create(&sock->sendThread, NULL, ntcp_thread_send, (void *)sock) != 0)
			DieWithError("create sendThread failed");
	}

	if(sock->recvThread == -1) {
		if( pthread_create(&sock->recvThread, NULL, ntcp_thread_recv, (void *)sock) != 0)
			DieWithError("create recvThread failed");	
	}
}


int 
ntcp_create_sock(struct ntcp_sock* sock, int ctrlsock, struct sockaddr_in* saddr, int src_port, int dstPort){
	sock->ctrlsock = ctrlsock;
	sock->src_port = src_port;
	sock->dst_port = dstPort;
	sock->srcAddr = saddr;
	sock->closed = false;
	sock->next_sent_seq = 0;
	sock->last_ack_seq = 0;
	sock->last_samp_ack = 0;
	sock->last_recv_sack_seq = 0;
	sock->last_sack_seq = 0;
	sock->last_timer_seq = 0;
	sock->recv_size = 0;
	sock->win_size_ = INIT_WIN_SIZE;
	sock->win_left_ = 0;
	sock->recv_win_left_ = 0;
	pthread_cond_init(&sock->cond_sendthread,NULL);
	pthread_mutex_init(&sock->mutex_sendthread, NULL);
	pthread_mutex_init(&sock->mutex_slidewin, NULL);
	pthread_mutex_init(&sock->mutex_sendque, NULL);
	pthread_mutex_init(&sock->mutex_total_sendque, NULL);
	pthread_mutex_init(&sock->mutex_tsque, NULL);
	pthread_mutex_init(&sock->mutex_sackmap, NULL);
	ntcp_init_timer(sock);
	
	sock->sendThread = -1;
	sock->recvThread = -1;
	sock->ack_ontime = true;
	sock->no_drop = true;
	sock->ack_wait = false;
	sock->sack_wait = false;
	sock->rtdata_timer_closed = true;
	sock->speed_adapter_on = false;
	
	sock->dupacks_ = 0;
	sock->srtt = 0;
	sock->rttvar = 0;
	sock->rto = 3000000;
	sock->total_recv = 0;
	sock->total_send = 0;
	sock->total_dup = 0;
	sock->total_sack = 0;
	sock->onetrip_send = 0;
	sock->onetrip_sack = 0;
	sock->dup_sack = 0;
	sock->cwnd_ = INIT_WIN_SIZE * SMSS;
	sock->ssthresh_ = INIT_WIN_SIZE * SMSS;
	
	gettimeofday(&sock->start_time, NULL);
	gettimeofday(&sock->last_sample_time, NULL);

	if ((sock->datasock = socket(PF_INET, SOCK_DGRAM, 0)) < 0) {
    		return FAILED;
	} else {
		return SUCCESS;
    }
}


int 
ntcp_recvfrom(struct ntcp_sock* sock, char* rcvBuffer, int size, int flag, struct sockaddr* addr, socklen_t* addr_len) 
{
	
	ntcp_init_threads(sock);
	int dataSize = 0;
	struct timespec req;
	req.tv_sec = 0;
	req.tv_nsec = 10000000;
	while(!sock->closed){
		if(sock->recv_queue.size() > 0){
			// recv a packet;
			struct ntcp_pkt* pkt = sock->recv_queue.front();
			sock->recv_queue.pop_front();
			dataSize = pkt->size;
			strncpy(rcvBuffer, pkt->rawdata, dataSize);
			break;
		}
		nanosleep(&req, NULL);
	};
	return dataSize;
}

int
ntcp_sendto(struct ntcp_sock* sock, char* sendBuffer, int size, int flag, struct sockaddr* addr, socklen_t addr_len)
{
	ntcp_init_threads(sock);
	
	struct timespec req;
	req.tv_sec = 0;
	req.tv_nsec = 2000000;

	// use cwnd_ to compare instead of win_size_
	while((sock->slide_win.size() * SMSS) >= (sock->cwnd_ /*+ sock->onetrip_sack*/)){
	/* 	if((sock->total_send > 0) && (sock->slide_win.size() < MAXWIN)){
			double bytes = sock->total_send * SMSS * 8.0/1024.0/1024.0;
			double during = get_time(sock);
			double speed = bytes/during;
			if(speed < 80){
				//double extra = (double)sock->total_send/speed*60.0;
				//printf("%f:averge speed %f extra pkt %d\n", during, speed, (int)extra);
				break;
			}
		}
	*/	nanosleep(&req, NULL);	
	};
	
	struct ntcp_pkt* pkt = new ntcp_pkt;
	memset(pkt->rawdata,0, MAXBUFSIZE);
	struct tcphdr *tcph = (struct tcphdr *)(pkt->rawdata);
	char* data = (char *)(pkt->rawdata + sizeof(struct tcphdr));	
 	
	tcph->source = htons(sock->src_port);
	tcph->dest = htons(sock->dst_port);
	tcph->seq = sock->next_sent_seq;
	sock->next_sent_seq += size;
	tcph->ack = 0;
	tcph->window =  sock->win_size_;
	strncpy(data,sendBuffer,size);
	pkt->size = size + sizeof(struct tcphdr);
	pkt->sacked = 0;
	pkt->seq = tcph->seq;

	ntcp_lock(sock, LOCK_SEND_QUEUE);
	ntcp_lock(sock, LOCK_SLIDE_WIN);
	ntcp_lock(sock, LOCK_TOTAL_SENQUE);

	sock->onetrip_send ++;
	sock->total_send ++;
	sock->dr_que[pkt->seq] = sock->total_send;
	sock->slide_win[pkt->seq] = pkt;
	sock->send_queue.push_back(pkt);

	ntcp_unlock(sock, LOCK_TOTAL_SENQUE);
	ntcp_unlock(sock, LOCK_SLIDE_WIN);
	ntcp_unlock(sock, LOCK_SEND_QUEUE);
	
	// notify thread to send data!
	ntcp_notify_sendthread(sock);

	return size;
}

void
ntcp_recvAckOrSack(struct ntcp_sock* sock, struct ntcp_pkt* pkt){
	
//	cout<<"ntcp_recv Ack or sack seq "<<pkt->seq<<endl;
	struct tcphdr* tcph = (struct tcphdr*)(pkt->rawdata);
	struct tcp_option* top = (struct tcp_option*)(pkt->rawdata + sizeof(struct tcphdr));
	map<unsigned int, struct ntcp_pkt*, less<unsigned int> >::iterator it, end;
	
	ntcp_stop_timer(sock,TIMER_RETRANSMIT_DATA);
	
	if(top->kind == 0){  // ack
		if(log_permit){
			printf("%f: ack received seq %d ack_seq %d rwnd %d win_size_ %d\n",get_time(sock),pkt->seq, tcph->ack_seq, tcph->window, sock->slide_win.size());
		}
		unsigned int be_seq = (*sock->slide_win.begin()).first; 
		if(tcph->ack_seq >= be_seq){
			
			// record last ack seq no
			sock->last_ack_seq = tcph->ack_seq;
			
			// compute rtt, rto
			ntcp_compute_rto(sock, pkt->seq);
		
			// lock slide window
			ntcp_lock(sock, LOCK_SLIDE_WIN);
		
			// adjust win_size_
			sock->total_recv += (tcph->ack_seq - be_seq)/(SENDBUFSIZE - 1);
			ntcp_adjust_cwnd(sock, tcph->window, pkt->seq);
				
			// slide send window
			it = sock->slide_win.begin();
			do{
				if((*it).first < tcph->ack_seq){
					sock->slide_win.erase(it ++);
				}else {
					break;
				}
			}while(it != sock->slide_win.end());
			
			sock->onetrip_sack = 0;
			ntcp_unlock(sock, LOCK_SLIDE_WIN);

		}else{
			//useless
		}

	}else if(top->kind == 5){ // sack
		if(log_permit){
			printf("%f: sack received with seq %d, sack length %d , win count %d\n",get_time(sock), tcph->seq, top->length, sock->slide_win.size());
		}
		sock->ack_ontime = false;
		sock->onetrip_sack ++;

		ntcp_clear_ts_que(sock, pkt->seq);
		
		unsigned int left_edge;
		unsigned int right_edge;
		unsigned int* edge = &top->edge0;

		ntcp_lock(sock,LOCK_SLIDE_WIN);

		if(sock->slide_win[pkt->seq]->sacked == 1){
		    sock->dup_sack ++;
		}
		
		for(int i = 0; i < top->length; i++){
			left_edge = edge[2*i];
			right_edge = edge[2*i + 1];
			sock->slide_win[left_edge]->sacked = 1;
//			printf(" %d: left %d right %d ",i,left_edge,right_edge);
			it = sock->slide_win.lower_bound(left_edge);
			
			while(it != sock->slide_win.end()){
				if((*it).second->seq >= right_edge) break;
				(*it).second->sacked = 1;
				it++;
			}
		}
		
		ntcp_unlock(sock, LOCK_SLIDE_WIN);
		sock->last_recv_sack_seq = pkt->seq;
	
		// check duplicated acks
		if(tcph->ack_seq == sock->last_ack_seq){
			ntcp_dup_actions(sock, pkt);
		}

	}else {
		// useless package
	}



	//reset retransmit data timer
	ntcp_reset_timer(sock, TIMER_RETRANSMIT_DATA);
}

void
ntcp_dup_actions(struct ntcp_sock* sock, struct ntcp_pkt* pkt) {
	
	map<unsigned int, struct ntcp_pkt*, less<unsigned int> >::iterator it, end;
	
	if(sock->state != STA_FAST_RECOVERY){
		sock->dupacks_ ++;

		// check fast recovery
		if(sock->dupacks_ == 3){
			sock->dupacks_ = 0;
			printf("%f FAST RECOVERY Begins\n", get_time(sock));
			// enter in fast recovery state 
			sock->state = STA_FAST_RECOVERY;

			ntcp_lock(sock,LOCK_SEND_QUEUE);
			ntcp_lock(sock,LOCK_SLIDE_WIN);
			ntcp_lock(sock,LOCK_TOTAL_SENQUE);
	
			// 1. Compute flightsize and ssthreshold
			sock->flight_size_ = 0;
			it = sock->slide_win.begin();
			while(it != sock->slide_win.end()){
				if((*it).second->sacked != 1) {
					sock->flight_size_ += (*it).second->size;
				}
				it ++;
			};
//			sock->flight_size_ = sock->slide_win.size() * SMSS;
			sock->ssthresh_ = double_max(sock->flight_size_/2.0, 2*SMSS);	
			
			// 2. Retransmit the lost segment and set cwnd to ssthresh plus 3*SMSS.
			sock->cwnd_ = sock->ssthresh_ + 3.0*SMSS;

			it = sock->slide_win.begin();
			std::deque<struct ntcp_pkt*> temp;
		
			int count = 0;
			while(it != sock->slide_win.end()) {
				if((*it).first >= pkt->seq) break;
				if((*it).second->sacked != 1){
					(*it).second->sacked = 2;
					temp.push_back((*it).second);
//					sock->send_queue.push_front((*it).second);
					count ++;
					sock->total_send ++;
					sock->onetrip_send ++;
					sock->dr_que[(*it).first] = sock->total_send;
				}
				it ++;
			};
			
			while(temp.size() > 0){
				sock->send_queue.push_front(temp.back());
				temp.pop_back();
			};

			printf("%f retrans %d pkt_\n", get_time(sock),count);
			ntcp_unlock(sock, LOCK_TOTAL_SENQUE);
			ntcp_unlock(sock, LOCK_SLIDE_WIN);
			ntcp_unlock(sock, LOCK_SEND_QUEUE);

			ntcp_notify_sendthread(sock);
		}
	}else{
		// 3. For each additional duplicate ACK received, increment cwnd by SMSS.
		sock->cwnd_ += SMSS;	
	}
}


void
ntcp_sendAck(struct ntcp_sock* sock) {

    	//compose ack packets
	struct ntcp_pkt* ackpkt = new ntcp_pkt;
	ackpkt->size = sizeof(struct tcphdr);
	memset(ackpkt->rawdata, 0, MAXBUFSIZE);
	struct tcphdr* tcph = (struct tcphdr*)(ackpkt->rawdata);
	tcph->source = htons(sock->src_port);
	tcph->dest = htons(sock->dst_port);
	tcph->seq = sock->wait_ack_seq;
	tcph->ack_seq = sock->win_left_; 
	tcph->ack = 1;
	tcph->window = sock->win_size_ - sock->recv_slide_win.size();
	struct tcp_option* sop = (struct tcp_option*)(ackpkt->rawdata + sizeof(struct tcphdr));
	sop->kind = 0;
	ntcp_lock(sock, LOCK_SEND_QUEUE);
	sock->send_queue.push_back(ackpkt);
	ntcp_unlock(sock, LOCK_SEND_QUEUE);
	if(log_permit){
		printf("%f: send ack_seq %d\n", get_time(sock),tcph->ack_seq);
	}
	// notify send thread
	ntcp_notify_sendthread(sock);
}


void
ntcp_sendSack(struct ntcp_sock* sock) {
	//compose sack packets
	struct ntcp_pkt* sack_pkt = new ntcp_pkt;
	sack_pkt->size = sizeof(struct tcphdr);
	memset(sack_pkt->rawdata, 0, MAXBUFSIZE);
	struct tcphdr* tcph = (struct tcphdr*)(sack_pkt->rawdata);
	tcph->source = htons(sock->src_port);
	tcph->dest = htons(sock->dst_port);
	tcph->seq = sock->last_sack_seq;
	tcph->ack_seq = sock->win_left_;
	tcph->ack = 1;
	struct tcp_option* sop = (struct tcp_option*)(sack_pkt->rawdata + sizeof(struct tcphdr));
	sop->kind = 5;
	unsigned int* edge = &sop->edge0;
	int count = 0;
	edge[0] = sock->last_sack_seq;
	edge[1] = sock->last_sack_seq + SENDBUFSIZE -1;
	count = 2;

/*	ntcp_lock(sock, LOCK_SACK_MAP);
	map<unsigned int, int, less<unsigned int> >::iterator cur = sock->sack_map.find(sock->last_sack_seq);
	if(cur == sock->sack_map.end()) {
		printf("last_sack_seq %d\n", sock->last_sack_seq);
		DieWithError("ntcp_sendSack(): no last_sack_seq find");
	}

	while((*cur).second != PKT_RIGHT_EDGE){
		cur ++;
		if(cur == sock->sack_map.end()){
		    DieWithError("ntcp_sendSack(): find right edge failed");
		}
	};
	int end_seq = (*cur).first;

	for(;; cur--){
	    switch((*cur).second){
		case PKT_LEFT_EDGE:
		    edge[count] = (*cur).first;
	//	    printf("trigger %d: left %d right %d\n", sock->last_sack_seq, (*cur).first,edge[count+1]);
		    count += 2;
		    break;
		case PKT_RIGHT_EDGE:	
		    edge[count+1] = (*cur).first;
		    break;
	    }
	    if(count == 4) break;
	    if(cur == sock->sack_map.begin()) break;
	}
	ntcp_unlock(sock, LOCK_SACK_MAP);
*/	sop->length = count/2;
	
	// essential to set pkt size
	sack_pkt->size += sizeof(struct tcp_option) + sizeof(int) * (count - 1);
	
	if(log_permit){
		printf("%f: last_sack_seq %d \n",get_time(sock), sock->last_sack_seq);
	}
	ntcp_lock(sock, LOCK_SEND_QUEUE);
	sock->send_queue.push_back(sack_pkt);
	ntcp_unlock(sock, LOCK_SEND_QUEUE);
	ntcp_notify_sendthread(sock);
}

void
ntcp_sendAckOrSack(struct ntcp_sock* sock,struct ntcp_pkt* pkt){
	
	if(log_permit){
		printf("%f: recv data seq %d\n",get_time(sock), pkt->seq);
	}
	if(sock->win_left_ == pkt->seq){
		// stop sack timer
		ntcp_stop_timer(sock, TIMER_DELAY_SACK);

		int nseq = 0;
		struct ntcp_pkt* next_pkt = pkt;
		map<unsigned int, struct ntcp_pkt*, less<unsigned int> >::iterator it;
		
		if(sock->ack_ontime)
			sock->recv_size ++;
		
		// add pkt into recv_slide_win
	    	sock->recv_slide_win[pkt->seq] = pkt;
		
		it = sock->recv_slide_win.find(pkt->seq);
		do{
			//clear sack area
			ntcp_remove_sack_area(sock,next_pkt);
			sock->recv_queue.push_back(next_pkt);
			sock->recv_slide_win.erase(it);
			sock->wait_ack_seq = next_pkt->seq;
			sock->win_left_ += next_pkt->size;
			it ++;
			if(it == sock->recv_slide_win.end()) break;
			else{
				next_pkt = (*it).second;
				nseq = next_pkt->seq;
			}
		}while(sock->win_left_ == nseq);
		
		if(sock->onetrip_sack != 0){
		    	printf(" %f: send sack count %d total_sack %ld total_dup %ld\n",get_time(sock), sock->onetrip_sack, sock->total_sack, sock->total_dup);
			sock->onetrip_sack = 0;
		}
		
		// if this pkt is retransmited, send ack now
		if(sock->ack_wait || (sock->last_sack_seq > pkt->seq)){
			sock->ack_wait = false;
			ntcp_stop_timer(sock, TIMER_DELAY_ACK);
			ntcp_sendAck(sock);
		}else{
			sock->ack_wait = true;
			ntcp_reset_timer(sock, TIMER_DELAY_ACK);
		}

	}else if (sock->win_left_ < pkt->seq){

		// add pkt into recv_slide_win
		sock->recv_slide_win[pkt->seq] = pkt;
		sock->onetrip_sack ++;
		sock->total_sack ++;

		//adjust sack area;
		ntcp_add_sack_area(sock, pkt);
		sock->last_sack_seq = pkt->seq;
		sock->no_drop = false;
		sock->ack_ontime = false;
		sock->recv_size ++;
		if(sock->ack_wait) {
			sock->ack_wait = false;
			ntcp_stop_timer(sock, TIMER_DELAY_ACK);
			ntcp_sendAck(sock);
		}
		ntcp_sendSack(sock);

	} else {
		sock->total_dup ++;
		// duplicate packets
		if(log_permit){
			printf("%f: dup pkt seq %d win_sack %ld total_dup %ld\n", get_time(sock),pkt->seq, sock->total_sack, sock->total_dup);
		}
	}
}

/* No use for this function */
double
ntcp_compute_drop_rate(struct ntcp_sock* sock)
{
	double win = 0;
	double lost = 0;
	deque<int>::iterator it = sock->win_que.begin();
	while(it != sock->win_que.end()){
		win += (*it);
		it ++;
	};
	it = sock->lost_que.begin();
	while(it != sock->lost_que.end()){
		lost += (*it);
		it ++;
	};
	return lost/win;

}

void
ntcp_clear_ts_que(struct ntcp_sock* sock, int last_seq)
{
	ntcp_lock(sock, LOCK_TIMESTAMP_QUEUE);
	while(sock->ts_que.size() > 0){
		if(sock->ts_que.front()->seq > last_seq) 
			break;
		delete sock->ts_que.front();
		sock->ts_que.pop();
	};
	ntcp_unlock(sock, LOCK_TIMESTAMP_QUEUE);
}

void 
ntcp_add_sack_area(struct ntcp_sock* sock, struct ntcp_pkt* pkt)
{
	int begin = pkt->seq;
	int end = begin + pkt->size;
//	printf("add_sack: begin %d end %d\n", begin, end);
	
//	ntcp_lock(sock, LOCK_SACK_MAP);
	switch(sock->sack_map[begin]){
	case 0:
		sock->sack_map[begin] = PKT_LEFT_EDGE;
		break;
	case PKT_RIGHT_EDGE:
		sock->sack_map[begin] = PKT_CENTER;
		break;
	}
	
	switch(sock->sack_map[end]){
	case 0:
		sock->sack_map[end] = PKT_RIGHT_EDGE;
		break;
	case PKT_LEFT_EDGE:
		sock->sack_map[end] = PKT_CENTER;
		break;
	}

//	ntcp_unlock(sock, LOCK_SACK_MAP);
}

void 
ntcp_remove_sack_area(struct ntcp_sock* sock, struct ntcp_pkt* pkt)
{
	int begin = pkt->seq;
	int end = begin + pkt->size;
//	printf("remove_sack: begin %d end %d\n", begin, end);
	
//	ntcp_lock(sock, LOCK_SACK_MAP);
	switch(sock->sack_map[begin]){
	case 0:
//		ntcp_unlock(sock,LOCK_SACK_MAP);
		return;
	case PKT_LEFT_EDGE:
		sock->sack_map.erase(begin);
		break;
	}
	
	switch(sock->sack_map[end]){
	case PKT_CENTER:
		sock->sack_map[end] = PKT_LEFT_EDGE;
		break;
	case PKT_RIGHT_EDGE:
		sock->sack_map.erase(end);
		break;
	}
//	ntcp_unlock(sock,LOCK_SACK_MAP);
}


void*
ntcp_thread_send(void* arg)
{
	int rc;
	int bytes;
	int remain;
	
	struct ntcp_sock* sock = (struct ntcp_sock*)arg;
	struct timespec wait_time;
	struct timeval sam_time;
	sam_time.tv_sec = 0;

	wait_time.tv_sec = 0;
	wait_time.tv_nsec = 8000000;
	cout<<"enter in ntcp_thread_send!"<<endl;
    	

	//start print tp timer
	ntcp_reset_timer(sock, TIMER_THROUGHPUT);

	for(;;){

		ntcp_lock(sock, LOCK_SEND_THREAD);

		while (!sock->send_queue.size()) {
			ntcp_unlock(sock, LOCK_SEND_QUEUE);
			//printf("Send Thread blocked\n");		
			if((rc = pthread_cond_wait(&sock->cond_sendthread, &sock->mutex_sendthread)) != 0){
				DieWithError("pthread_cond_wait failed");
			}

			ntcp_lock(sock, LOCK_SEND_QUEUE);	
		}

		// Release lock
		ntcp_unlock(sock, LOCK_SEND_THREAD);


		struct ntcp_pkt* pkt = sock->send_queue.front();
		if(pkt == 0){
		   	 ntcp_unlock(sock, LOCK_SEND_QUEUE);
			 continue;
		}
		int len = pkt->size;
		struct tcphdr *tcph = (struct tcphdr *)(pkt->rawdata);
		
		if(tcph->ack == 0) {
			if(pkt->sacked != 2){
			    struct ntcp_timestamp* ts = new ntcp_timestamp;
			    ts->seq = pkt->seq;
			    if(gettimeofday(&ts->timestamp, NULL) < 0)
				DieWithError("ntcp_thread_send(): gettimeofday() failed");
			    ntcp_lock(sock, LOCK_TIMESTAMP_QUEUE);
			    sock->ts_que.push(ts);
			    ntcp_unlock(sock, LOCK_TIMESTAMP_QUEUE);
			}
		}
		
		if(log_permit){
			printf("%f: send pkt seq %d \n",get_time(sock),tcph->seq);
		}
			/* Send data */
	    	if ((bytes = sendto(sock->datasock, pkt->rawdata, len, MSG_CONFIRM, (const struct sockaddr*)sock->srcAddr, sizeof(struct sockaddr_in))) != len){
			if( (bytes == -1) || (bytes <= sizeof(struct tcphdr))){
				//printf("***bytes %d\n", bytes);
				nanosleep(&wait_time, NULL);
				ntcp_unlock(sock, LOCK_SEND_QUEUE);
				continue;
				//DieWithError("resend data!");
			}else{			
				DieWithError("send problem!\n");
			}
		}

		if( (tcph->ack == 0) ) {
			if(sock->rtdata_timer_closed){
				// start retransmit timer
				ntcp_reset_timer(sock, TIMER_RETRANSMIT_DATA);
				sock->last_timer_seq = pkt->seq;
			}
		}
		sock->send_queue.pop_front();

		//sched_yield();

	}
	return NULL;
}


void*
ntcp_thread_recv(void *arg)
{
	struct ntcp_sock* sock = (struct ntcp_sock*)arg;
	char buf[MAXBUFSIZE];
	struct sockaddr_in* srcAddr = sock->srcAddr;
	struct tcphdr* tcph = (struct tcphdr*)buf;
	char* data = (char*)(buf + sizeof(struct tcphdr));
	int recvMsgSize = 0;
	int dataSize = 0;
	
	unsigned int rcvPort = htons(sock->src_port);
	
	cout<<"recv thread running"<<endl;

	socklen_t addr_len = sizeof(struct sockaddr_in);
	while(!sock->closed)
	{
		memset(buf, 0, MAXBUFSIZE);
		if ((recvMsgSize = recvfrom(sock->datasock, buf, MAXBUFSIZE-1 , 0, (struct sockaddr*)sock->srcAddr, &addr_len)) <= 0){
    			printf("recvMsgSize %d\n", recvMsgSize);
		//	continue;
			DieWithError("ntcp_thread_recv() - recvfrom() failed");
		}
	//	cout<<"recv msg size "<<recvMsgSize<<" port "<<ntohs(tcph->dest)<< " need "<<ntohs(rcvPort)<<endl;
		if(rcvPort == tcph->dest){
		    if((dataSize=recvMsgSize - sizeof(struct tcphdr))>=0){ 
			
			struct ntcp_pkt* pkt = new ntcp_pkt;
			memset(pkt->rawdata, 0, MAXBUFSIZE);
			pkt->seq = tcph->seq;
			if(tcph->ack == 0){
			    
			    // set rwnd
			    sock->win_size_ = tcph->window;
			    
			    pkt->size = dataSize;
			    memcpy(pkt->rawdata,data,pkt->size);
			    ntcp_sendAckOrSack(sock, pkt);
			}else {
			    pkt->size = recvMsgSize;
			    memcpy(pkt->rawdata, buf, pkt->size);
			    ntcp_recvAckOrSack(sock, pkt);		
			}
		    }
		}
//		sched_yield();
   	};
}

void
ntcp_lock(struct ntcp_sock* sock, int locktype){
	pthread_mutex_t* mutex;
	int rc;

	switch(locktype){
	case LOCK_SEND_THREAD:
		mutex = &sock->mutex_sendthread;
		break;
	case LOCK_SLIDE_WIN:
		mutex = &sock->mutex_slidewin;
		break;
	case LOCK_SEND_QUEUE:
		mutex = &sock->mutex_sendque;
		break;
	case LOCK_TOTAL_SENQUE:
		mutex = &sock->mutex_total_sendque;
		break;
	case LOCK_TIMESTAMP_QUEUE:
		mutex = &sock->mutex_tsque;
		break;
	case LOCK_SACK_MAP:
		mutex = &sock->mutex_sackmap;
		break;
	}
	if((rc=pthread_mutex_lock(mutex)) != 0)
		DieWithError("ntcp_lock(): pthread_mutex_lock() failed");
    }

void
ntcp_unlock(struct ntcp_sock* sock, int locktype){
	pthread_mutex_t* mt;
	int rc;

	switch(locktype){
		case LOCK_SEND_THREAD:
			mt = &sock->mutex_sendthread;
			break;
	    case LOCK_SLIDE_WIN:
			mt = &sock->mutex_slidewin;
			break;
	    case LOCK_SEND_QUEUE:
			mt = &sock->mutex_sendque;
			break;
	    case LOCK_TOTAL_SENQUE:
	    	mt = &sock->mutex_total_sendque;
	    	break;
	    case LOCK_TIMESTAMP_QUEUE:
	    	mt = &sock->mutex_tsque;
			break;
	    case LOCK_SACK_MAP:
			mt = &sock->mutex_sackmap;
			break;
	}
	if((rc = pthread_mutex_unlock(mt)) != 0)
	    DieWithError("ntcp_unlock(): pthread_mutex_unlock() failed");
}

void
ntcp_notify_sendthread(struct ntcp_sock* sock)
{
	ntcp_lock(sock, LOCK_SEND_THREAD);

	if(pthread_cond_signal(&sock->cond_sendthread) != 0)
		DieWithError("notifySendFunc(): pthread_cond_signal() failed");

	ntcp_unlock(sock, LOCK_SEND_THREAD);	
//	printf("notify send thread to send\n");
}

void
ntcp_reset_timer(struct ntcp_sock* sock, int type)
{
	struct itimerspec its; 
	switch(type){
	case TIMER_DELAY_ACK:
		if(timer_settime(sock->timer_da, 0, &its_da, NULL) == -1)
			DieWithError("start ack retransmit timer failed");
		break;
	case TIMER_DELAY_SACK:
		if(timer_settime(sock->timer_dsack, 0, &its_dsack, NULL) == -1)
			DieWithError("start data retransmit timer failed");
		break;
	case TIMER_RETRANSMIT_DATA:
		sock->rtdata_timer_closed = false;
		its.it_value.tv_sec = sock->rto / 1000000;
		its.it_value.tv_nsec = sock->rto % 1000000 * 1000;
		its.it_interval.tv_sec = 0;
		its.it_interval.tv_nsec = 0;
		if(timer_settime(sock->timer_rt_data, 0, &its, NULL) == -1)
			DieWithError("start data retransmit timer failed");
		break;
	case TIMER_THROUGHPUT:
		if(timer_settime(sock->timer_tp, 0 , &its_tp, NULL) == -1)
			DieWithError("start throughput timer failed");
		break;
	}
}

void
ntcp_stop_timer(struct ntcp_sock* sock, int type)
{
	timer_t otimer;
	switch(type){
	case TIMER_DELAY_SACK:
		otimer = sock->timer_dsack;
		break;
	case TIMER_DELAY_ACK:
		otimer = sock->timer_da;
		break;
	case TIMER_RETRANSMIT_DATA:
		sock->rtdata_timer_closed = true;
		otimer = sock->timer_rt_data;
		break;
	case TIMER_THROUGHPUT:
		otimer = sock->timer_tp;
		break;
	}
	if(timer_settime(otimer, 0, &its_empty, NULL) == -1)
	    DieWithError("reset timer failed");
}

void
ntcp_sendSack_handler(int sig, siginfo_t *si, void* parm)
{
	struct ntcp_sock* sock = (struct ntcp_sock*)si->si_value.sival_ptr;
	ntcp_sendSack(sock);
}

void
ntcp_sendAck_handler(int sig, siginfo_t *si, void* parm)
{
	struct ntcp_sock* sock = (struct ntcp_sock*)si->si_value.sival_ptr;
	ntcp_sendAck(sock);
}


void
ntcp_print_throughput_handler(int sig, siginfo_t *si, void* parm)
{
	struct ntcp_sock* sock = (struct ntcp_sock*)si->si_value.sival_ptr;
	double bytes = sock->last_ack_seq - sock->last_samp_ack;	
	if(sock->last_ack_seq == 0){
		ntcp_stop_timer(sock, TIMER_THROUGHPUT);
		return;
	}
	double tp = (bytes * 8.0 / 1024.0 /1024.0)/ (double)SAMPLE_INV * 1000.0;
	printf("%f %f %f\n", get_time(sock), tp, sock->cwnd_);
	sock->last_samp_ack = sock->last_ack_seq;
}

void
ntcp_retransmitData_handler(int sig, siginfo_t *si, void* parm)
{
	struct ntcp_sock* sock = (struct ntcp_sock*)si->si_value.sival_ptr;
	
	int rt_count = 0;
	int last_seq = -1;
	ntcp_lock(sock, LOCK_SEND_QUEUE);
	ntcp_lock(sock, LOCK_SLIDE_WIN);
	ntcp_lock(sock, LOCK_TOTAL_SENQUE);
	
	map<unsigned int, struct ntcp_pkt*, less<unsigned int> >::iterator it = sock->slide_win.begin();
	int orig_size = sock->send_queue.size();
	while(it != sock->slide_win.end()){
		if((*it).second->sacked != 1){
			(*it).second->sacked = 2;
			sock->send_queue.push_back((*it).second);
			sock->total_send ++;
			sock->onetrip_send ++;
			sock->dr_que[(*it).first] = sock->total_send;
			rt_count ++;
			//printf("%f:retransmit seq %d\n",get_time(sock),(*it).first);
		}
		last_seq = (*it).first;
		it ++;
	};
	ntcp_unlock(sock, LOCK_TOTAL_SENQUE);
	ntcp_unlock(sock, LOCK_SLIDE_WIN);
	ntcp_unlock(sock, LOCK_SEND_QUEUE);

	// avoid empty retransmit
	if(last_seq == -1) return;

	// clear rtt timestamp queue
	ntcp_clear_ts_que(sock, last_seq);
	
	// back off RTO
	sock->ack_ontime = false;
	if(sock->rto * 1.2 > MAX_RTO) {
		sock->rto = MAX_RTO;
	} else{
		sock->rto *= 1.2;
	}

	printf("%f: retransmit %d pkt, backoff rto %ld win count %d,received sack %d, ssthresh_ %f \n",get_time(sock),rt_count, sock->rto, sock->slide_win.size(), sock->onetrip_sack, sock->ssthresh_);
	sock->onetrip_sack = 0;

	ntcp_notify_sendthread(sock);
}

void
ntcp_adjust_cwnd(struct ntcp_sock* sock, int rwnd, int pkt_seq)
{
	if(sock->state == STA_FAST_RECOVERY) {
		sock->state = STA_CONGESTION_AVOID;
		// When the next ACK arrives that acknowledges new data, set cwnd to ssthresh
		printf("%f: Exit Fast Recovery\n", get_time(sock));
		sock->cwnd_ = sock->ssthresh_;
		//sock->cwnd_ = (sock->cwnd_ + sock->ssthresh_)/2.0;

	}else{
		// compute cwnd
		ntcp_compute_cwnd(sock, rwnd, pkt_seq);
	}

	// change it to the win size
	sock->win_size_ = (int)(sock->cwnd_ / (double)SMSS);
	//if(sock->win_size_ < INIT_WIN_SIZE)
	//	sock->win_size_ = INIT_WIN_SIZE;
}


void
ntcp_compute_cwnd(struct ntcp_sock* sock, int rwnd, int pkt_seq)
{
	if(sock->dr_que.size() == 0) return;

	// check if slow start
	if(sock->cwnd_ < sock->ssthresh_){
		// increase at most SMSS
		sock->cwnd_ += SMSS;
//		sock->cwnd_ *= 2.0;
		return;
	}

	// keep sample interval to be RTT
	struct timeval sample_ts;
	gettimeofday(&sample_ts, NULL);
	long during = (sample_ts.tv_sec - sock->last_sample_time.tv_sec) * 1000000 + (sample_ts.tv_usec - sock->last_sample_time.tv_usec);
	if(during < sock->rto) return;
	else{
		sock->last_sample_time.tv_sec = sample_ts.tv_sec;
		sock->last_sample_time.tv_usec = sample_ts.tv_usec;
	}
	
	ntcp_lock(sock, LOCK_TOTAL_SENQUE);
	// calculate package drop rate
	map<unsigned int, long, std::less<unsigned int> >::iterator it = sock->dr_que.begin();
	do{
		if((*it).first < pkt_seq){
			sock->dr_que.erase(it++);
		} else if((*it).first > pkt_seq) {
			printf("ntcp_adjust_cwnd(): find total_send failed\n");
			break;
		} else{
			break;	
		}
	}while(it != sock->dr_que.end());

	double drop_rate = (1.0 - (double)sock->total_recv/(double)sock->dr_que[pkt_seq] );
	ntcp_unlock(sock, LOCK_TOTAL_SENQUE);
	
	double incr = (double)SMSS*SMSS/sock->cwnd_;
	double factor = (sock->cwnd_ > 40000)?1.1:30;
	sock->cwnd_ += (( incr > 1.0)?incr:1.0)*factor;
	//printf("%f: new cwnd_ %f , drop rate %f\n", get_time(sock), sock->cwnd_, drop_rate);
}

void
ntcp_compute_rto(struct ntcp_sock* sock, int tg_seq)
{
	struct timeval rtt_stop;
	double R = 0;
	int G = 20* 1000; // 20 ms now
	int K = 4;
	double beta = 0.25;
	double alpha = 0.125;
	struct timeval* rtt_start;
	
	if(sock->ts_que.size() == 0) return;
	
	ntcp_lock(sock, LOCK_TIMESTAMP_QUEUE);
	if( sock->ts_que.front()->seq  <= tg_seq) {
	
		int last_que_seq = 0;
		while(sock->ts_que.front()->seq < tg_seq)
		{
			last_que_seq = sock->ts_que.front()->seq;
			sock->ts_que.pop();
			if(sock->ts_que.size() == 0){
				printf("ntcp_compute_rto(): timestamp queue is empty tg_seq %d last_que_seq %d\n",tg_seq, last_que_seq);
				ntcp_unlock(sock, LOCK_TIMESTAMP_QUEUE);
				return;
			}
		};
		rtt_start = &sock->ts_que.front()->timestamp;
		sock->ts_que.pop();

		if(gettimeofday(&rtt_stop, NULL) < 0)
			DieWithError("ntcp_compute_rto(): gettimeofday() failed");
	
		R = (rtt_stop.tv_sec - rtt_start->tv_sec)* 1000000 + rtt_stop.tv_usec - rtt_start->tv_usec;
		if(R < 0){
			printf("start sec %ld stop sec %ld R %f\n", rtt_start->tv_sec, rtt_stop.tv_sec, R);
			DieWithError("Rtt is negtive");
		}
			
    		if(sock->srtt == 0){
			//SRTT <- R
			//RTTVAR <- R/2
			sock->srtt = R;
			sock->rttvar = R/2.0;
		}else{
			//RTTVAR <- (1 - beta) * RTTVAR + beta * |SRTT - R'|
			//SRTT <- (1 - alpha) * SRTT + alpha * R'
			sock->rttvar = (1.0-beta) * sock->rttvar + beta * fabs(sock->srtt - R);
			sock->srtt = (1.0 - alpha) * sock->srtt + alpha * R;
		}
		// RTO <- SRTT + max (G, K*RTTVAR)
		sock->rto = (long)(sock->srtt + int_max(G, (int)K*sock->rttvar));
	//	printf("%f:srtt %f, rttvar %f, rto %ld, R-rtt %f win count %d, seq %d\n", get_time(sock), sock->srtt, sock->rttvar, sock->rto, R, sock->slide_win.size(), tg_seq);
	}
	ntcp_unlock(sock, LOCK_TIMESTAMP_QUEUE);
}

void 
ntcp_close(struct ntcp_sock* sock)
{
	//select and recv
	fd_set fdsr;
	int maxsock = sock->ctrlsock;
	int ret;	
	struct timeval tv;
	
	if(sock->ack_wait){
		ntcp_stop_timer(sock, TIMER_DELAY_ACK);
		ntcp_sendAck(sock);
		sock->ack_wait = false;
	}

	while( sock->recv_slide_win.size() != 0)
	{
		FD_ZERO(&fdsr);
		FD_SET(sock->ctrlsock, &fdsr);
		tv.tv_sec = 1;
		tv.tv_usec = 0;

		ret = select(maxsock + 1, &fdsr, NULL, NULL, &tv);
		if(ret < 0){
			break;
		}
		ntcp_sendAck(sock);
		sleep(2);
	};
	while( sock->slide_win.size() != 0 )
	{
		sleep(2);
	};
	close(sock->ctrlsock);
}

